package com.atguigu.api

import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
 * @ClassName ProcessFunctionTest
 * @Description 某段时间内温度连续上升，报警提示
 * @Author Mr Yang
 * @Date 2020/9/2 22:57
 * @Version 1.0
 */
object ProcessFunctionTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream = env.socketTextStream("localhost", 7777)
    //transform
    val dataStream = stream.map(
      data => {
        val dataArray = data.split(",")
        SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
      })
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
        override def extractTimestamp(t: SensorReading): Long = t.timestamp * 1000
      })

    val processStream = dataStream.keyBy(_.id)
      .process(new TemperatuerIncreAlert())

    processStream.print("processStream")
    dataStream.print("dataStream")
    env.execute("window test")
  }
}

class TemperatuerIncreAlert() extends KeyedProcessFunction[String, SensorReading, String] {

  //定义一个状态，用来保存上一个数据的温度值
  lazy val lastTemp: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemp",classOf[Double]))

  //定义一个状态，用来保存定时器的时间戳
  lazy val currentTimestamp: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("currentTimestamp", classOf[Long]))


  override def processElement(i: SensorReading, context: KeyedProcessFunction[String, SensorReading, String]#Context, collector: Collector[String]): Unit = {
    //先取出上一个温度值
    val perTemp = lastTemp.value()
    //更新温度值
    lastTemp.update(i.temperature)

    val currentTimerTs = currentTimestamp.value()

    //温度连上升且没有注册定时器，则注册定时器
    if (i.temperature > perTemp && currentTimerTs == 0){
      //获取当前时间
      val currentTime = context.timerService().currentProcessingTime()
      //注册定时器
      context.timerService().registerProcessingTimeTimer(currentTime + 10000L)

      currentTimestamp.update(currentTimerTs)
    }else if (perTemp > i.temperature || perTemp == 0.0){
      //删除定时器并清空状态
      context.timerService().deleteProcessingTimeTimer(currentTimerTs)
      currentTimestamp.clear()
    }
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {
    //输出报警信息
    out.collect(ctx.getCurrentKey + " 温度连续上升")
    currentTimestamp.clear()
  }
}